package drds.binlog.metadata;

import com.google.common.base.Function;
import com.google.common.collect.ComputingMap;
import com.google.common.collect.Lists;
import com.google.common.collect.MapMaker;
import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.position.Position;
import drds.binlog.common.position.PositionRange;

import java.util.List;
import java.util.Map;

/**
 * 内存版实现
 */
public class MemoryMetaDataManager extends AbstractLifeCycle implements MetaDataManager
{

    protected Map<String, List<ClientInfo>> destinationToClientInfoListMap;
    protected Map<ClientInfo, Position> clientInfoToPositionMap;
    protected Map<ClientInfo, BatchIdToPositionRangeConcurrentMap> clientInfoToBatchIdToPositionRangeMapMap;


    public void start()
    {
        super.start();
        //
        destinationToClientInfoListMap = ComputingMap.makeComputingMap(new Function<String, List<ClientInfo>>()
        {

            public List<ClientInfo> apply(String destination)
            {
                return Lists.newArrayList();
            }
        });

        clientInfoToPositionMap = new MapMaker().makeMap();
        clientInfoToBatchIdToPositionRangeMapMap = ComputingMap.makeComputingMap(new Function<ClientInfo, BatchIdToPositionRangeConcurrentMap>()
        {

            public BatchIdToPositionRangeConcurrentMap apply(ClientInfo clientInfo)
            {
                return BatchIdToPositionRangeConcurrentMap.create(clientInfo);
            }

        });

    }

    public void stop()
    {
        super.stop();
        //
        destinationToClientInfoListMap.clear();
        clientInfoToPositionMap.clear();
        for (BatchIdToPositionRangeConcurrentMap batchIdToPositionRangeConcurrentMap : clientInfoToBatchIdToPositionRangeMapMap.values())
        {
            batchIdToPositionRangeConcurrentMap.clear();
        }
    }

    //
    public synchronized void subscribe(ClientInfo clientInfo) throws MetaDataManagerException
    {
        List<ClientInfo> clientInfoList = destinationToClientInfoListMap.get(clientInfo.getDestination());

        if (clientInfoList.contains(clientInfo))
        {
            clientInfoList.remove(clientInfo);
        }
        clientInfoList.add(clientInfo);
    }

    public synchronized boolean hasSubscribed(ClientInfo clientInfo) throws MetaDataManagerException
    {
        List<ClientInfo> clientInfoList = destinationToClientInfoListMap.get(clientInfo.getDestination());
        return clientInfoList != null && clientInfoList.contains(clientInfo);
    }

    public synchronized void unsubscribe(ClientInfo clientInfo) throws MetaDataManagerException
    {
        List<ClientInfo> clientInfoList = destinationToClientInfoListMap.get(clientInfo.getDestination());
        if (clientInfoList != null && clientInfoList.contains(clientInfo))
        {
            clientInfoList.remove(clientInfo);
        }
    }

    public synchronized List<ClientInfo> getClientInfoList(String destination) throws MetaDataManagerException
    {
        return Lists.newArrayList(destinationToClientInfoListMap.get(destination));
    }

    //
    public void updatePosition(ClientInfo clientInfo, Position position) throws MetaDataManagerException
    {
        clientInfoToPositionMap.put(clientInfo, position);
    }

    public Position getPosition(ClientInfo clientInfo) throws MetaDataManagerException
    {
        return clientInfoToPositionMap.get(clientInfo);
    }
    //

    public Long addPositionRange(ClientInfo clientInfo, PositionRange positionRange) throws MetaDataManagerException
    {
        return clientInfoToBatchIdToPositionRangeMapMap.get(clientInfo).addPositionRange(positionRange);
    }

    // 添加记录到指定batchId
    public void addPositionRange(ClientInfo clientInfo, Long batchId, PositionRange positionRange) throws MetaDataManagerException
    {
        clientInfoToBatchIdToPositionRangeMapMap.get(clientInfo).addPositionRange(batchId, positionRange);
    }

    public PositionRange removePositionRange(ClientInfo clientInfo, Long batchId) throws MetaDataManagerException
    {
        return clientInfoToBatchIdToPositionRangeMapMap.get(clientInfo).removePositionRange(batchId);
    }

    //
    public PositionRange getPositionRange(ClientInfo clientInfo, Long batchId) throws MetaDataManagerException
    {
        return clientInfoToBatchIdToPositionRangeMapMap.get(clientInfo).getPositionRange(batchId);
    }

    public PositionRange getFirstPositionRange(ClientInfo clientInfo) throws MetaDataManagerException
    {
        return clientInfoToBatchIdToPositionRangeMapMap.get(clientInfo).getFirstPositionRange();
    }

    public PositionRange getLastestPositionRange(ClientInfo clientInfo) throws MetaDataManagerException
    {
        return clientInfoToBatchIdToPositionRangeMapMap.get(clientInfo).getLastestPositionRange();
    }

    public void clearBatchIdToPositionRangeMap(ClientInfo clientInfo) throws MetaDataManagerException
    {
        clientInfoToBatchIdToPositionRangeMapMap.get(clientInfo).clear();
    }

    public Map<Long, PositionRange> getBatchIdToPositionRangeMap(ClientInfo clientInfo) throws MetaDataManagerException
    {
        return clientInfoToBatchIdToPositionRangeMapMap.get(clientInfo).getBatchIdToPositionRangeMap();
    }


}
